package com.rocketmq.demo;

import com.rocketmq.constants.MqConstants;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.junit.Test;

import java.util.UUID;

/**
 * @Author 派大星
 * @Description TODO
 * @Date 2025/7/28 14:50
 */
public class HKeyTest {

    /**
     * key 一般代表的是业务参数 我们从业务上面要保证key的唯一性
     *
     * @throws Exception
     */
    @Test
    public void tagProducer() throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("key-producer-group");
        producer.setNamesrvAddr(MqConstants.NAMESERVER_ADDR);
        producer.start();
        String key = UUID.randomUUID().toString();
        System.out.println(key);
        // 创建一个带tag的消息
        Message message1 = new Message("keyTopic", "vip1", key, "我是vip1的文章".getBytes());
        producer.send(message1);
        System.out.println("发送成功");
        producer.shutdown();
    }

    @Test
    public void keyConsumer() throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("key-consumer-group");
        consumer.setNamesrvAddr(MqConstants.NAMESERVER_ADDR);
        consumer.subscribe("keyTopic", "*");
        consumer.registerMessageListener((MessageListenerConcurrently) (messageExtList, context) -> {
            MessageExt messageExt = messageExtList.get(0);
            System.out.println("MessageKey: " + messageExt.getKeys());
            System.out.println("我是VIP1的消费者 消费消息：" + new String(messageExtList.get(0).getBody()));
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        consumer.start();
        System.in.read();
    }
}
